package process;

import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.TimestampAssigner;
import org.apache.flink.api.common.eventtime.TimestampAssignerSupplier;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.ArrayList;
import java.util.List;

/**
 * Event Time 定时器
 */
public class EventTimeTimerDemo {

    public static void main(String[] args) throws Exception{
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());

        DataStreamSource<String> lines = env.socketTextStream("localhost", 8888);

        // 提取watermark
        SingleOutputStreamOperator<String> watermarks = lines.assignTimestampsAndWatermarks(WatermarkStrategy.<String>forBoundedOutOfOrderness(Duration.ofMillis(0))
                .withTimestampAssigner((s, ts) -> Long.parseLong(s.split(",")[0])));


        SingleOutputStreamOperator<Tuple2<String, Integer>> mapped = watermarks.map(new MapFunction<String, Tuple2<String, Integer>>() {
            @Override
            public Tuple2<String, Integer> map(String s) throws Exception {
                String[] fields = s.split(",");
                return Tuple2.of(fields[1], Integer.parseInt(fields[2]));
            }
        });


        // KeyBy
        KeyedStream<Tuple2<String, Integer>, String> keyed = mapped.keyBy(f -> f.f0);

        // process
        keyed.process(new KeyEventTimeProcess()).print();
        env.execute("");
    }

    private static class KeyEventTimeProcess extends KeyedProcessFunction<String,Tuple2<String,Integer>, Tuple2<String,Integer>>{

        // 状态
        private transient ValueState<List<Tuple2<String,Integer>>> state;

        @Override
        public void open(Configuration parameters) throws Exception {
            ValueStateDescriptor<List<Tuple2<String, Integer>>> stateDescriptor = new ValueStateDescriptor<List<Tuple2<String, Integer>>>("lst_state", TypeInformation.of(new TypeHint<List<Tuple2<String, Integer>>>() {}));
            state = getRuntimeContext().getState(stateDescriptor);
        }

        @Override
        public void onTimer(long timestamp, OnTimerContext ctx, Collector<Tuple2<String, Integer>> out) throws Exception {
            System.out.println("on timer : " + timestamp);
            List<Tuple2<String, Integer>> lst = state.value();
            for (Tuple2<String, Integer> t : lst) {
                out.collect(t);
            }
            state.update(null);
        }



        @Override
        public void processElement(Tuple2<String, Integer> value,Context context, Collector<Tuple2<String, Integer>> collector) throws Exception {
            List<Tuple2<String, Integer>> lst = state.value();
            if (lst == null){
                lst = new ArrayList<>();
            }
            lst.add(value);
            state.update(lst);
            long watermark = context.timerService().currentWatermark();
            System.out.println("currentWatermark: " + watermark);
            context.timerService().registerEventTimeTimer(watermark + 5000L);
            System.out.println("register timer: " + (watermark + 5000L));
        }
    }

}
